package com.atguigu.bigdata.chapter05.sink;

import com.alibaba.fastjson.JSON;
import com.atguigu.bigdata.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

/**
 * @Author lzc
 * @Date 2022/9/3 9:16
 */
public class Flink03_Sink_Redis {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
    
    
        FlinkJedisPoolConfig redisConf = new FlinkJedisPoolConfig.Builder()
            .setHost("hadoop162")
            .setPort(6379)
//            .setDatabase(1) // 选择数据库
            .setMaxTotal(100)// 连接提供的最大连接数
            .setMaxIdle(10)  // 最大空闲连接数
            .setMinIdle(2)  // 最小空闲连接数
            .build();
    
        env
            .readTextFile("input/sensor.txt")
            .map(new MapFunction<String, WaterSensor>() {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(
                        data[0],
                        Long.valueOf(data[1]),
                        Integer.valueOf(data[2])
                    );
                }
            })
            // select id, sum(vc) from t group by id;
            .keyBy(WaterSensor::getId)  // 方法引用
            .sum("vc")
            .addSink(new RedisSink<>(redisConf, new RedisMapper<WaterSensor>() {
                
                // 返回一个命令描述符   字符串: set list: rpush lpush ...
                @Override
                public RedisCommandDescription getCommandDescription() {
                    // 参数1: 只能命令(确定要写的数据类型)
                    // 参数2: 额外的key, 当写出的的类型是 hash和zset的时候有效
                    return new RedisCommandDescription(RedisCommand.SET, null);
                }
    
                // 从数据中返回key. 返回的这条数据写入的时候的key
                @Override
                public String getKeyFromData(WaterSensor data) {
                    // 用sensor id 作为每条数据的key
                    return data.getId();
                }
    
                // 返回写入的具体数据
                @Override
                public String getValueFromData(WaterSensor data) {
                    return JSON.toJSONString(data);
                }
            }));
            
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
/*
redis的数据结构:
string

list

set

hash

zset


 */